package org.mijia.dotsbox;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.mijia.dotsbox.task.Task;
import org.mijia.dotsbox.task.TaskEventListener;
import org.mijia.dotsbox.utils.LogUtils;

public class TaskDispatcher implements TaskEventListener {

	private HashSet<String> pendingTasks = new HashSet<String>();
	private ExecutorService worker = null;
	private ScheduledExecutorService timer = null;
	private List<TaskEventListener> taskListeners = new ArrayList<TaskEventListener>();
	
	public TaskDispatcher() {
		LogUtils.info("Initializing DotsBox Task Dispatcher");
		
		// init services
		int nThreadCount = Runtime.getRuntime().availableProcessors() * 2 + 1;
		LogUtils.debug("DispatcherThreadPool: CorePoolSize=" + nThreadCount);
		if (worker == null) {
			worker = new DispatcherThreadPool(nThreadCount);
		}
		if (timer == null) {
			timer = Executors.newSingleThreadScheduledExecutor();
		}
		
		addEventListener(this);
	}
	
	public void dispose() {
		if (worker != null) {
            worker.shutdown();
        }

        if (timer != null) {
            timer.shutdown();
        }

        LogUtils.info("DotsBox Task Scheduler Disposed");
	}
	
	public void execute(Task task) {
		boolean bContain = false;
		synchronized (pendingTasks) {
			bContain = pendingTasks.contains(task.getId());
		}
		if (bContain)
			return;
		
		if (task.isTimerTask()) {
			timer.scheduleWithFixedDelay(task, task.getDelaySeconds(), task
					.getPeriodSeconds(), TimeUnit.SECONDS);
		} else {
			worker.execute(task);
		}
		synchronized (pendingTasks) {
			pendingTasks.add(task.getId());
		}
	}
	
	public void addEventListener(TaskEventListener listener) {
		if (!taskListeners.contains(listener))
			taskListeners.add(listener);
	}
	
	private void beforeExecuteTask(Task task) {
		for (TaskEventListener listener : taskListeners) {
			listener.taskStartExecute(task);
		}
	}
	
	private void afterExecuteTask(Task task) {
		synchronized (pendingTasks) {
			pendingTasks.remove(task.getId());
        }

		for (TaskEventListener listener : taskListeners) {
			listener.taskEndExecute(task);
		}
	}
	
	// task event listener implementation
	public void taskEndExecute(Task task) {
		StringBuffer sb = new StringBuffer();
		sb.append("[END] ").append(task.getStatus()).append(" [").append(
				task.getCost()).append(" ms]");
		LogUtils.info(sb.toString());
	}

	public void taskStartExecute(Task task) {
		StringBuffer sb = new StringBuffer();
		sb.append("[BEGIN] ").append(task.getStatus());
		LogUtils.info(sb.toString());
	}
	
	private class DispatcherThreadPool extends ThreadPoolExecutor {
		
		public DispatcherThreadPool(int nThread) {
			super(nThread, nThread, 0L, TimeUnit.MILLISECONDS,
					new LinkedBlockingQueue<Runnable>());
		}

		@Override
		protected void beforeExecute(Thread t, Runnable r) {
			if (r instanceof Task) {
				beforeExecuteTask((Task) r);
			}
			super.beforeExecute(t, r);
		}
		
		@Override
		protected void afterExecute(Runnable r, Throwable t) {
			super.afterExecute(r, t);
			if (r instanceof Task) {
				// ignore the uncaught exception of *t*
				afterExecuteTask((Task) r);
				if (t != null) {
					LogUtils.warning("There is something happend in task dispatcher: " + t.getMessage());
				}
			}
		}

	}

}
